package org.suyafei.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.suyafei.func.AsyncLocaltionRequestFunction;
import org.suyafei.func.DeviceMapFunction;
import org.suyafei.func.LocationMapFunction;
import org.suyafei.model.DeviceInfoDO;
import org.suyafei.model.ShortLinkWideDO;
import org.suyafei.util.DeviceUtil;
import org.suyafei.util.KafkaUtil;

import java.util.concurrent.TimeUnit;

/**
 * @author 宿某人
 * @Date: 2022/03/31/22:58
 * @Description:
 */
@Slf4j
public class DwmShortLinkWideApp {

    /**
     * 定义topic
     */
    public static final String SOURCE_TOPIC = "dwd_link_visit_topic";

    /**
     * 消费者组
     */
    public static final String GROUP_ID = "dwm_short_link_group";


    public static final String SINK_TOPIC = "dwm_link_visit_topic";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(SOURCE_TOPIC, GROUP_ID);

        DataStreamSource<String> ds = env.addSource(kafkaConsumer);

        //格式转换，补齐设备信息
        SingleOutputStreamOperator<ShortLinkWideDO> deviceWideDS = ds.map(new DeviceMapFunction());
        deviceWideDS.print("设备新奇宽表补齐");

        SingleOutputStreamOperator<String> locationDS = AsyncDataStream.orderedWait(deviceWideDS, new AsyncLocaltionRequestFunction(), 1000, TimeUnit.MILLISECONDS, 100);
//        SingleOutputStreamOperator<String> locationDS = deviceWideDS.map(new LocationMapFunction());
        locationDS.print("地理位置信息补齐");

        //存储到dwd
        FlinkKafkaProducer<String> kafkaProducer = KafkaUtil.getKafkaProducer(SINK_TOPIC);
        locationDS.addSink(kafkaProducer);
        env.execute();

    }
}
